package com.atguigu.flink.chapter05.Source;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/8/9 10:02
 */
public class Flink05_MySource {
    public static void main(String[] args) throws Exception {
        // 0.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        MySource mySource = new MySource();

        DataStreamSource<WaterSensor> myDS = env.addSource(mySource);

        SingleOutputStreamOperator<WaterSensor> resultDS = myDS.map(new MapFunction<WaterSensor, WaterSensor>() {
            @Override
            public WaterSensor map(WaterSensor value) throws Exception {
                if (value.getVc() < 10) {
                    mySource.cancel();
                }
                return value;
            }
        });

        resultDS.print();


        env.execute();
    }


    public static class MySource implements SourceFunction<WaterSensor> {

        private static boolean isRunning = true;

        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            Random random = new Random();
            while (isRunning) {
                ctx.collect(new WaterSensor(
                        "sensor_" + random.nextInt(3),
                        System.currentTimeMillis(),
                        random.nextInt(100)
                ));
                Thread.sleep(1000);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
